package com.atguigu.app.func;

import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.TableProcess;
import com.atguigu.common.EduConfig;
import com.atguigu.utils.DruidDSUtil;
import com.atguigu.utils.JdbcUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

/**
 * ClassName: DimTableProcessFunction
 * Package: com.atguigu.app.func
 * Description:
 *
 * @Author 尚硅谷-李增亮
 * @Create 2023/5/15 16:38
 * @Version 1.0
 */
public class DimTableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private MapStateDescriptor<String, TableProcess> stateDescriptor;

    private HashMap<String, TableProcess> tableProcessHashMap;


    public DimTableProcessFunction(MapStateDescriptor<String, TableProcess> stateDescriptor) {
        this.stateDescriptor = stateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //  使用JDBC直接读取配置信息表,将数据存入HashMap
        tableProcessHashMap = new HashMap<>();


        //  使用JDBC连接数据库
        Connection connection = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/edu_config?" + "user=root" +
                "&password=000000&useUnicode=true&" + "characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");


        //  数据库查询、并转化：下划线转化为驼峰
        List<TableProcess> queryList = JdbcUtil.queryList(connection,
                "select  * from table_process where sink_type='dim'",
                TableProcess.class,
                true);

        //  通过checkTable方法、连接Phoenix 并建表
        for (TableProcess tableProcess : queryList) {
            checkTable(tableProcess);
            tableProcessHashMap.put(tableProcess.getSourceTable(), tableProcess);
        }

        System.out.println("预加载。。。。");

        //  关闭JDBC连接
        connection.close();


    }

    //  通过预编译语句和 Phoenix 连接对象来执行建表操作，根据表格信息生成建表语句，并使用预编译语句执行建表操作。
    private void checkTable(TableProcess tableProcess) {

        //获取连接
        DruidPooledConnection phoenixConn = null;
        PreparedStatement preparedStatement = null;

        //获取建表相关字段
        String sinkTable = tableProcess.getSinkTable();
        String sinkPk = tableProcess.getSinkPk();
        String sinkColumns = tableProcess.getSinkColumns();
        String sinkExtend = tableProcess.getSinkExtend();

        try {
            //  获取Phoenix的连接对象
            phoenixConn = DruidDSUtil.getPhoenixConn();

            //封装建表语句 create table if not exists db.sinkTable(id varchar primary key,name varchar,sex varchar) xxx
            //  如果没有主键、则给默认值 ： id 作为主键
            if (sinkPk == null || "".equals(sinkPk)) {
                sinkPk = "id";
            }
            //  如果没有扩展字段、默认空字符串
            if (sinkExtend == null) {
                sinkExtend = "";
            }

            //  建表语句 写入到Phoenix中、创建对应表格
            StringBuilder createTableSql = new StringBuilder("create table if not exists ")
                    .append(EduConfig.HBASE_SCHEMA)
                    .append(".")
                    .append(sinkTable)
                    .append("(");
            //  将sinkColumns切分成个例、获取主键
            String[] columnArr = sinkColumns.split(",");
            for (int i = 0; i < columnArr.length; i++) {
                String column = columnArr[i];
                if (sinkPk.equals(column)) {
                    createTableSql.append(column).append(" varchar primary key");
                } else {
                    createTableSql.append(column).append(" varchar");
                }

                //  如果不是最后一个列 以”,"分割否则以”)“结尾
                if (i < columnArr.length - 1) {
                    createTableSql.append(",");
                } else {
                    createTableSql.append(")").append(sinkExtend);
                }
            }

            //  打印SQL语句
            System.out.println("建表语句为：" + createTableSql);

            //  预编译SQL
            preparedStatement = phoenixConn.prepareStatement(createTableSql.toString());

            //  执行建表操作
            preparedStatement.execute();

        } catch (SQLException e) {
            e.printStackTrace();
            throw new RuntimeException("建表" + sinkTable + "失败!");
        } finally {
            //  归还连接
            if (preparedStatement != null) {
                try {
                    preparedStatement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
            //  归还连接
            if (phoenixConn != null) {
                try {
                    phoenixConn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }


    // TODO {"before":{"id":2,"category_name":"aaaa","create_time":null,"update_time":null,"deleted":null},"after":{"id":2,"category_name":"cccc","create_time":null,"update_time":null,"deleted":null},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1684203100000,"snapshot":"false","db":"edu","sequence":null,"table":"base_category_info","server_id":1,"gtid":null,"file":"mysql-bin.000007","pos":705,"row":0,"thread":null,"query":null},"op":"u","ts_ms":1684203099972,"transaction":null}
    //  处理配置流的数据
    @Override
    public void processBroadcastElement(String value, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
        //  因为需要将数据放入状态里面、所以需要转化为JavaBean对象

        //  获取广播状态、并将状态描述器传入 、初始化广播变量
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(stateDescriptor);

        //  解析value的格式、并转化为JSONObject
        JSONObject jsonObject = JSONObject.parseObject(value);
        //  如果为删除操作,则删除状态中的数据
        if ("d".equals(jsonObject.getString("op"))) {
            //  使用 JSONObject.parseObject 方法将 JSON 字符串转换为 TableProcess 对象
            TableProcess tableProcess = JSONObject.parseObject(jsonObject.getString("before"), TableProcess.class);
            //  从 TableProcess 对象中获取源表名 key。
            String key = tableProcess.getSourceTable();
            //  同步更新表配置信息。
            broadcastState.remove(key);
            //  除对应的表配置信息。
            tableProcessHashMap.remove(key);
        } else {
            //  将数据转换为TableProcess对象
            TableProcess tableProcess = JSONObject.parseObject(jsonObject.getString("after"), TableProcess.class);

            //  增加判断，防止dwd层数据进入
            if ("dim".equals(tableProcess.getSinkType())) {
                //  使用checkTable方法：校验并在Phoenix建表
                checkTable(tableProcess);

                //  将数据写入状态
                broadcastState.put(tableProcess.getSourceTable(), tableProcess);
            }



        }

    }


    //  设计要求 {"database":"edu","table":"order_info","type":"insert","ts":1684198335,"xid":34620,"commit":true,"data":{"id":20813,"user_id":47,"origin_amount":200.00,"coupon_reduce":0.00,"final_amount":200.00,"order_status":"1001","out_trade_no":"226273892319854","trade_body":"Redis等1件商品","session_id":"c6ca231f-ec93-44b0-9215-f24178f75ea4","province_id":21,"create_time":"2023-05-16 08:52:15","expire_time":"2023-05-16 09:07:15","update_time":null}}
    //  配置主流的数据信息
    @Override
    public void processElement(JSONObject value, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //  获取状态信息
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(stateDescriptor);

        //  获取table
        String table = value.getString("table");
        //  在状态中查找对应的表配置
        TableProcess tableProcess = broadcastState.get(table);
        //  从本地中获取相应的表配置
        TableProcess mapTableProcess = tableProcessHashMap.get(table);

        //  insert update delete bootstrap-start bootstrap-insert bootstrap-complete
        String type = value.getString("type");

        //  根据输入数据的类型判断：是否要进行行过滤和列过滤
        //  行过滤
        if ((tableProcess != null || mapTableProcess != null) && ("insert".equals(type) || "update".equals(type) || "bootstrap-insert".equals(type))) {

            //  如果表只存在本地中而且状态中没有 、 将本地赋值给状态使用
            if (tableProcess == null) {
                tableProcess = mapTableProcess;
            }

            //2.列过滤
            filterColumns(value.getJSONObject("data"), tableProcess.getSinkColumns());

            //3.写出
            value.put("sink_table", tableProcess.getSinkTable());
            value.put("pk",tableProcess.getSinkPk());
            out.collect(value);

        } else {
            System.out.println("该表:" + table + "没有对应的配置信息！或者类型不匹配:" + type);
        }


    }

    /**
     * 列过滤
     *
     * @param data        {"id":"1001","name":"zs","sex":"male","tm_name":"aaa"}
     * @param sinkColumns id,tm_name
     */
    private void filterColumns(JSONObject data, String sinkColumns) {

        //将sinkColumns切分并转换为集合
        String[] columns = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columns);

//        Set<Map.Entry<String, Object>> entries = data.entrySet();
//        Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
//        while (iterator.hasNext()) {
//            Map.Entry<String, Object> next = iterator.next();
//            if (!columnList.contains(next.getKey())) {
//                iterator.remove();
//            }
//        }

        Set<Map.Entry<String, Object>> entries = data.entrySet();
        entries.removeIf(next -> !columnList.contains(next.getKey()));
    }

}
